#ifndef __M_MSG_H__
#define __M_MSG_H__
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <string>
#include <iostream>
#include<list>
#include <unordered_map>
#include <memory>
namespace bitmq
{
    using MessagePtr = std::shared_ptr<bitmq::MQMessage>;
    #define DATAFILE_SUBFIX ".mqd"
    #define TMPFILE_SUBFIX ".mqd.tmp"
    class MessageMapper//消息持久的映射
    {
        public:
        MessageMapper(std::string basedir,const std::string&qname):_qname(qname){
         
            if(basedir.back()!='/')basedir.push_back('/');
            _datafile=basedir+qname+DATAFILE_SUBFIX;
            _tmpfile=basedir+qname+TMPFILE_SUBFIX;
            if(FileHelper(basedir).exits()==false)
            {
                assert(FileHelper::createDirectory(basedir));
            }
         
            createMsgFile();
           
        }
        //创建存储数据文件
        bool createMsgFile(){
            if(FileHelper(_datafile).exits()==true)return true;
            bool ret=FileHelper::createFile(_datafile);
            if(!ret)
            {
                DLOG("创建数据文件失败：%s",_datafile.c_str());
                return false;
            }
            return true;
        }
        //移除数据文件和临时文件
        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }
        bool insert(MessagePtr&ptr)
        {
            return insert(_datafile,ptr);
        }
        //移除数据文件的指定消息   思路:是将该消息的有效数据标志置"0",再将消息数据，覆盖式的写入到文件
        bool remove(MessagePtr&ptr)
        {
             //1、将有效标志置  "0"
             //ptr->payload()  这个不能修改数据，返回的式const 对象
             ptr->mutable_payload()->set_valid("0");
             //2、将消息序列化后，判断
             std::string body=ptr->payload().SerializeAsString();
             FileHelper file(_datafile);
            
             size_t body_size=body.size();
            if(body_size!=ptr->length())
            {
                DLOG("序列化后字符串的长度与实际消息数据长度不符!");
                return false;
            }
             //3、写入文件
             bool ret=file.write(body.c_str(),ptr->offset(),body_size);
             if(!ret)
             {
                DLOG("消息写入文件失败！");
                return false;
             }
             return true;
        }
        std::list<MessagePtr> gc()
        {
            std::list<MessagePtr> result;
            //1、将加载文件中所有的有效数据
            bool ret=load(result);
            if(!ret)
            {
                DLOG("加载有效数据失败！");
                return result;
            }
            std::cout<<"加载历史数据完成"<<std::endl;
            //2、再将有效数据写入临时文件中
            FileHelper::createFile(_tmpfile);
            for(auto&e:result)
            {
                DLOG("向临时文件写入数据: %s", e->payload().body().c_str());
                ret=insert(_tmpfile,e);
                if(!ret)
                {
                    DLOG("数据写入临时文件失败！");
                    return result;
                }
            }
            //3、删除源文件
            
            ret=FileHelper::removeFile(_datafile);
            if(!ret)
            {
                DLOG("删除源文件失败!");
                return result;
            }
            //4、将临时文件改成数据文件
            ret=FileHelper(_tmpfile).rename(_datafile);
            if(!ret)
            {
                DLOG("修改临时文件失败!");
                return result;
            }
            //5、返回有效 数据
            return result;
        }
        private:
        //加载数据文件中所有有效数据进list<Messageptr>   思路：先取出消息长度   再取出消息   再继续往后走
        bool load(std::list<MessagePtr>&result)
        {
            bitmq::FileHelper data_file(_datafile);
            size_t offset=0,msg_size;
            size_t data_file_size=data_file.size();

            bool ret;
            while(offset<data_file_size)
            {
                //先读取出消息长度
                ret=data_file.read((char*)&msg_size,offset,sizeof(size_t));
                if(!ret)
                {
                    DLOG("读取文件大小失败！");
                    return false;
                }
                offset+= sizeof(size_t);

                std::string msg_body(msg_size,'\0');
                //再取出消息数据
                ret=data_file.read((char*)&msg_body[0],offset,msg_size);
                if(!ret)
                {
                    DLOG("读取消息数据失败");
                    return false;
                }
                offset+=msg_size;
                //构造对象
                MessagePtr msg_ptr=std::make_shared<MQMessage>();
                //将消息数据进行反序列化
                msg_ptr->mutable_payload()->ParseFromString(msg_body);
                if(msg_ptr->payload().valid()=="0")
                {
                    DLOG("加载到无效消息：%s", msg_ptr->payload().body().c_str());
                    continue;
                }
                result.push_back(msg_ptr);
            }
            return true;
        }
        //向指定的文件末尾添加数据
        bool insert(const std::string&filename,MessagePtr&msg_ptr)
        {
            //1、将文件序列化，再向文件中添加指定的格式   4字节|数据|4字节|数据
            std::string body=msg_ptr->payload().SerializeAsString();
            //2、获取文件大小
            FileHelper file(filename);
            size_t file_len=file.size();//文件偏移量
            size_t body_size=body.size();//要写入文件数据的大小
            //3、向文件写入数据  1、先写4字节（数据长度） 2、再写数据
            //std::cout<<"开始写入文件为: "<<body<<std::endl;
            bool ret=file.write((char*)&body_size,file_len,sizeof(size_t));
            if(!ret)
            {
                DLOG("写入4字节的文件大小失败!--%s",filename.c_str());
                return false;
            }

            ret=file.write(body.c_str(),file_len+sizeof(size_t),body_size);
            if(!rand){
                DLOG("写入文件数据失败!--%s",filename.c_str());
                return false;
            }

            //4、更新实际消息的偏移量（文件起始位置）
            msg_ptr->set_offset(file_len+sizeof(size_t));
            msg_ptr->set_length(body_size);
            return true;  
        }
        private:
        std::string _qname;
        std::string _datafile;
        std::string _tmpfile;
    };

    class QueueMessage{
        public:
        using ptr=std::shared_ptr<QueueMessage>;
        QueueMessage(const std::string&basedir,const std::string&qname):_map(basedir,qname),_qname(qname),_valid_count(0),_total_count(0){
        }
        bool recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _wait_push=_map.gc();
            for(auto&msg:_wait_push)
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
            }
            _total_count=_valid_count=_wait_push.size();
            return true;
        }
        bool insert( BasicProperties*bp,const std::string&body,bool queue_is_mode)//插入消息
        {
            //1、构造对象
            MessagePtr msg=std::make_shared<bitmq::MQMessage>();
            msg->mutable_payload()->set_body(body);
            if(bp!=nullptr)
            {
                DeliveryMode mode=queue_is_mode?bp->delivery_mode():DeliveryMode::UNDURABLE;//
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode=queue_is_mode?DeliveryMode::DURABLE:DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            std::unique_lock<std::mutex> lock(_mutex);
            //2、判断是否消息持久化
            if(msg->payload().properties().delivery_mode()==DeliveryMode::DURABLE)
            {
                //3、插入数据
                msg->mutable_payload()->set_valid("1");
                bool ret=_map.insert(msg);
                if(!ret)
                {
                    DLOG("插入持久化消息失败！%s",body.c_str());
                    return false;
                }
        
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
                _valid_count+=1;
                _total_count+=1;
            }
            //4、内存管理
            _wait_push.push_back(msg);
            return true;
        }
        bool remove(const std::string&msg_id)//每次删除持久化消息，都判断一下，要不要回收垃圾
        {   
            //1、判断消息是否在待确认hash里
            std::unique_lock<std::mutex> lock(_mutex);
            auto it=_wait_acks.find(msg_id);
            if(it==_wait_acks.end())
            {
                DLOG("没有找到要删除的消息！%s",msg_id.c_str());
                return true;
            }            
            //2、判断是否是持久化数据
            if(it->second->payload().properties().delivery_mode()==DeliveryMode::DURABLE)
            {
                //3、删除持久化数据
                bool ret=_map.remove(it->second);
                if(!ret)
                {
                    DLOG("删除持久化消息失败!");
                    return false;
                }
                _valid_count-=1;
                _durable_msgs.erase(msg_id);
                gc();//每次删除都判断一次要不要，回收垃圾数据
            }
            
            //4、删除待确认ack里消息的数据
            _wait_acks.erase(msg_id);
            return true;
        }
        MessagePtr front()//获取队首消息
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if(_wait_push.size()==0)//判断待推送链表是否为空
            {
                return MessagePtr();
            }
            MessagePtr msg=_wait_push.front();
            _wait_push.pop_front();

            _wait_acks.insert(std::make_pair(msg->payload().properties().id(),msg));//要将消息添加到待确认应答hash
            return msg;
        }
        size_t total_count() {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        size_t getable_count()//获取待推送消息的数量
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _wait_push.size();
        }
        size_t wait_ack_count()//获取待ack的数量
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _wait_acks.size();
        }
        size_t durable_count()//获取持久化消息的数量
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        void clear()//清理所有的消息
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _map.removeMsgFile();
            _wait_push.clear();
            _durable_msgs.clear();
            _wait_acks.clear();
            _valid_count=0;
            _total_count=0;
        }
        private:
        bool GCChack()
        {
            if(_total_count>2000&&_valid_count*10/_total_count>5)
            {
                return true;
            }
            return false;
        }
        void gc()//回收垃圾消息
        {
            if(!GCChack())return;

            //1、获取所有有效消息的链表
            std::list<MessagePtr> msg=_map.gc();
            for(auto&e:msg)
            {
                auto it=_durable_msgs.find(e->payload().properties().id());
                if(it==_durable_msgs.end())
                {
                    DLOG("在内存中没找到持久化数据！");
                    _wait_push.push_back(e);
                    _durable_msgs.insert(std::make_pair(e->payload().properties().id(),e));
                    continue;
                }
                //2、更新回收后消息的实际位置
                it->second->set_offset(e->offset());
                it->second->set_length(e->length());
                
            }
            _valid_count=_total_count=msg.size();
        }
        private:
        std::mutex _mutex;
        std::string _qname;//队列名称
        size_t _valid_count;//有效消息个数
        size_t _total_count;//消息的chocho
        MessageMapper _map;//消息的句柄
        std::list<MessagePtr> _wait_push;//待推送消息链表
        std::unordered_map<std::string,MessagePtr> _durable_msgs;//持久化消息的hash
        std::unordered_map<std::string,MessagePtr> _wait_acks;//待确认的消息的hash
       
    };

//向外提供的接口  消息的管理
    class MessageManager{
        public:
        using ptr=std::shared_ptr<MessageManager>;
        MessageManager( std::string basedir):_basedir(basedir){}
        void clear()//清理所有队列消息
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto&msg:_queue_msgs)
            {
                msg.second->clear();
            }
        }
        void initQueueMessage(const std::string&qname)//初始化队列
        {   
            
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_queue_msgs.find(qname);
                if(it!=_queue_msgs.end())return;
                qmp=std::make_shared<QueueMessage>(_basedir,qname);
              
                _queue_msgs.insert(std::make_pair(qname,qmp));
            }
            qmp->recovery();
        }
        void destroyQueueMessage(const std::string&qname)//销毁一个队列消息
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_queue_msgs.find(qname);
                if(it==_queue_msgs.end())return;

                qmp=it->second;
                _queue_msgs.erase(qname);
            }
            qmp->clear();
        }
        MessagePtr getFrontMessage(const std::string&qname)//获取队列头部消息
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_queue_msgs.find(qname);
                if(it==_queue_msgs.end())
                {
                    DLOG("获取队列%s队首消息失败:没有找到消息管理句柄!",qname.c_str());
                    return MessagePtr();
                }
                qmp=it->second;
            }
            return qmp->front();
        }
        bool insert(const std::string&qname,bitmq::BasicProperties*bp,const std::string&body,bool mode)//向队列里添加数据
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_queue_msgs.find(qname);
                if(it==_queue_msgs.end())
                {
                    DLOG("向队列%s添加消息失败,没找消息管理句柄",qname.c_str());
                    return false;
                }

                qmp=it->second;
            }
            return qmp->insert(bp,body,mode);
        }
        bool ack(const std::string& qname,const std::string&msg_id)//对消息进行确认
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_queue_msgs.find(qname);
                if(it==_queue_msgs.end())
                {
                    DLOG("对消息确认失败,没找到消息管理句柄%s",qname.c_str());
                    return false;
                }

                qmp=it->second;
            }
            return qmp->remove(msg_id);
        }
        size_t getable_count(const std::string&qname)//获取待推送消息的数量
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_queue_msgs.find(qname);
                if(it==_queue_msgs.end())
                {
                    DLOG("获取待推送消息的数量,没找到消息管理句柄%s",qname.c_str());
                    return false;
                }

                qmp=it->second;
            }
            return qmp->getable_count();
        }
        size_t durable_count(const std::string&qname)//获持久化消息的数量
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_queue_msgs.find(qname);
                if(it==_queue_msgs.end())
                {
                    DLOG("获持久化消息的数量,没找到消息管理句柄%s",qname.c_str());
                    return false;
                }

                qmp=it->second;
            }
            return qmp->durable_count();
        }
        size_t ack_count(const std::string&qname)//待ack确认的消息数量
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_queue_msgs.find(qname);
                if(it==_queue_msgs.end())
                {
                    DLOG("待ack确认的消息数量,没找到消息管理句柄%s",qname.c_str());
                    return false;
                }

                qmp=it->second;
            }
            return qmp->wait_ack_count();
        }
        size_t total_count(const std::string&qname)//总的消息的数量
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_queue_msgs.find(qname);
                if(it==_queue_msgs.end())
                {
                    DLOG("总的消息的数量,没找到消息管理句柄%s",qname.c_str());
                    return false;
                }

                qmp=it->second;
            }
            return qmp->total_count();
        }
        private:
        std::mutex _mutex;
        std::string _basedir;//
        std::unordered_map<std::string,QueueMessage::ptr> _queue_msgs;//队列名称&持久化队列消息句柄
    };
}
#endif